package com.feiwei

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.types.Row

object day9_TableTimeAndWindow {


  def main(args: Array[String]): Unit = {



    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    environment.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)

    //1.11 直接创建时blink'
    val builder = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
        .build()

    val tableEnv= StreamTableEnvironment.create(environment,builder)

    //从文件中读取流
    val path="E:\\repository\\company\\myself\\flink-learning\\flink-learning-demo\\src\\main\\resources\\sensor.txt"

    val lineStream = environment.readTextFile(path)

     var stream=  lineStream.map(v=>{

      val arr = v.split(",")

      SensorReading(arr(0),arr(1).toLong,arr(2).toDouble)
    })

    //从流中转换成table
    //在最后指定 proctime
    var t1=tableEnv.fromDataStream(stream,'id,'temperature,'timestamp,'ps.proctime)


    t1.printSchema()

    t1.toAppendStream[Row].print()


    environment.execute()

  }




}
